feat(reporting): notify the adopter when a reporting issue opens, escalates or resolves - #1875

Summary

Closes #1258. A seller can now learn, through a supported callback, when a reporting issue opens, escalates, is acknowledged or waived, or resolves. ESA can drop its get_reporting_status response scan.

from adcp.reporting import ReliableReportingService, ReportingIssueEvent

async def on_issue(event: ReportingIssueEvent) -> None:
    if event.issue is not None and event.issue.responsible_party != "buyer":
        await page(event)  # dedupe on event.event_id

ReliableReportingService(..., issue_observer=on_issue)

Stacked on the storage PR (#1874): until that merges, this diff also shows its commit. Merge #1874 first; this one merges after the adcp-reporting-storage beta release that ships migration 0008. Against a store without the capability, the SDK refuses issue_observer with an actionable error that names adcp-reporting-storage 0.2.0b4.

API

• issue_observer: ReportingIssueObserver on ReliableReportingService. It is any sync or async callable taking a ReportingIssueEvent.
• ReportingIssueEvent (frozen) has:
  • event_id: stable across redeliveries.
  • kind: opened | escalated | acknowledged | waived | resolved.
  • occurred_at, account_id, consumer_id, issue_id, attempt.
  • issue: the ReportingIssue as get_reporting_status rendered it for that caller at occurred_at, with responsible_party, recommended_action, severity, opened_at and the causing reporting_status_id. It is None for resolved.
  • reporting_status_id, a convenience accessor.
• ReportingIssueEventDispatcher for adopters who compose the ledger without the service.
• Ledger level:
  • ReportingIssueEventStore, the capability protocol, plus supports_issue_events().
  • ReportingIssueEventLease and issue_event_id().
  • The in-memory reference store implements all of it.
• ReliableReportingTurn.issue_events reports escalations_recorded, delivered and failed.

When it fires

• After commit, once per transition. The store records each event in the same transaction as its lifecycle write: ingest settle, read settle, set_issue_state, retirement and replay persistence. The record is exactly once across replicas.
• On the worker turn. run_worker() or worker_interval delivers committed events. The same turn sweeps escalations. It settles the account first, so a cleared condition resolves instead of escalating, and an unattended statement escalates on time without anyone polling.
• At least once. A crash between the observer returning and the delivery stamp redelivers the event with the same event_id. Deduplicate on it. This is documented in the module docstring, the service docstring and the guide.
• In order. An occurrence's events arrive in recording order, at most one leased at a time.
• Never into the buyer's call.
  • An observer that raises, or an async observer past 10 s, is reported to worker_error_handler as issue_observer:{event_id}.
  • The event is retried after a backoff of 30 s doubling to 1 h.
  • A transient snapshot-read failure is retried rather than delivered unrendered.
• Idempotent. An exact replay of sync_reporting_status (unchanged) and repeated status reads record nothing.

Scope decisions

• production=. Leaving this out was agreed. The service raises ReliableReportingConfigurationError when production= and issue_observer are passed together. Follow-up: feat(reporting): support issue_observer on the production= service path #1873.
• Rendering. The issue is re-projected from the account snapshot as of occurred_at: statements recorded by then (chain leaf as of then), obligations and revisions created by then, and the occurrence in the event's state. Revision readability is mutable and has no history, so an event delivered after the seller changed readability can render issue=None. Rendering is documented as best effort.
• TS parity. @adcp/sdk has no issue-level hook; its closest analog is the obligation-health subscribers/notify outbox. This design mirrors that model: a durable record, at-least-once delivery, a stable id, retry until success, a 10 s timeout. A follow-up asks TS for issueObserver with the same event_id derivation: feat(reporting): issueObserver — notify the seller when a reporting issue opens, escalates or resolves (Python parity) adcp-client#3648.

Tests

• tests/test_reporting_issue_observer.py (memory):
  • the service flow: open on ingest, one event; replay, status read and idle turn record nothing; escalation at the boundary only, once; resolution on agreement
  • a failing observer: reported, retried with the same event_id, attempt=2
  • per-occurrence ordering behind a failed event
  • timeout of a hung async observer; sync observer
  • operator acknowledged and waived events rendering in their state
  • a cleared condition resolves rather than escalates
  • refusals: production=, an unsupported store, a non-callable observer, a missing table
  • event_id parity with the storage distribution
• tests/conformance/reporting/test_reporting_issue_observer_pg.py runs the same flows on postgres_storage, plus two replicas delivering each event once and refusal before migration 0008.
• tests/type_checks/reporting_issue_observer.py (mypy --strict and pyright).

Docs: "Observe reporting issues" in docs/reliable-reporting-service.md, and the worker_error_handler component list.

🤖 Generated with Claude Code